package cn.com.lyb.flink.aggreagte;

import cn.com.lyb.flink.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 有了按键分区的数据流KeyedStream，我们就可以基于它进行聚合操作了。Flink为我们内置实现了一些最基本、最简单的聚合API，主要有以下几种：
 * sum()：在输入流上，对指定的字段做叠加求和的操作。
 * min()：在输入流上，对指定的字段求最小值。
 * max()：在输入流上，对指定的字段求最大值。
 * minBy()：与min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而minBy()则会返回包含字段最小值的整条数据。
 * maxBy()：与max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。
 * 简单聚合算子使用非常方便，语义也非常明确。这些聚合方法调用时，也需要传入参数；但并不像基本转换算子那样需要实现自定义函数，只要说明聚合指定的字段就可以了。指定字段的方式有两种：指定位置，和指定名称。
 */
public class SimpleAggreagteDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 3),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3),
                new WaterSensor("s3", 3L, 3));

        KeyedStream<WaterSensor, String> sebsorkKS = sensorDS.keyBy((KeySelector<WaterSensor, String>) value -> value.getId());

        sebsorkKS.sum("ts").print();

        System.out.println("===========");
        sebsorkKS.max("vc").print();

        System.out.println("===========");
        sebsorkKS.min("ts").print(); // 取了比较字段的最小值，非比较字段保留第一次的值

        System.out.println("===========");
        sebsorkKS.minBy("vc").print();// 取了比较字段的最小值，同时非比较字段 取 最小值的这条数据的值
        env.execute();
    }
}
